package State

import Source.SensorReading
import org.apache.flink.api.common.functions.{ReduceFunction, RichMapFunction}
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, MapState, MapStateDescriptor, ReducingState, ReducingStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._

import java.util

object KVState {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    env.setParallelism(1)

    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })


    env.execute("state test")

  }
}

/**
 * Keyed state测试：必须定义在RichFunction中，因为需要运行时上下文
 */
class MyRichMapper extends RichMapFunction[SensorReading, String] {

  var valueState: ValueState[Double] = _
  lazy val listState: ListState[Int] = getRuntimeContext.getListState(
    new ListStateDescriptor[Int]("listState", classOf[Int]))

  lazy val mapState: MapState[String, Double] = getRuntimeContext.getMapState(
    new MapStateDescriptor[String, Double]("mapState", classOf[String], classOf[Double])
  )

  lazy val reduceState: ReducingState[SensorReading] = getRuntimeContext.getReducingState(
    new ReducingStateDescriptor[SensorReading]("reduceState",
      new ReduceFunction[SensorReading] {
        override def reduce(t: _root_.Source.SensorReading, t1: _root_.Source.SensorReading) = ???
      }, classOf[SensorReading])
  )

  override def open(parameters: _root_.org.apache.flink.configuration.Configuration): Unit = {
    valueState = getRuntimeContext.getState(
      new ValueStateDescriptor[Double]("valueState", classOf[Double]))
  }


  override def map(in: _root_.Source.SensorReading): _root_.scala.Predef.String = {
    //状态读取数据
    val myValueState = valueState.value()
    //状态更改值
    valueState.update(in.temperature)
    //ListState的赋值
    listState.add(1)
    val list = new util.ArrayList[Int]()
    list.add(1)
    list.add(2)
    //直接将列表的数据全部赋值
    listState.addAll(list)
    //listState的更新,将里面的值全部换成list里的
    listState.update(list)
    //是否包含
    mapState.contains("sensor_1")
    //更新数据
    mapState.put("sensor_1", 1.0)
    //获取值
    mapState.get("sensor_1")


    //获取聚合完成的值
    reduceState.get()
    //这是再加一个数据和已经聚合处理好的数据再次聚合处理
    reduceState.add(in)

    in.id
  }
}